package notify

import (
	"errors"
	"fmt"
	"iot-base/common/common"
	"iot-base/common/config"
	"iot-base/common/ctrltype"
	"iot-base/common/eventtype"
	"iot-base/common/innermq"
	"iot-base/common/logger"
	"iot-base/common/params"
	"iot-base/common/propertytype"
	"iot-base/common/redis"
	"iot-base/common/types"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"

	redis2 "github.com/go-redis/redis/v8"
	"github.com/vmihailenco/msgpack/v5"
)

var (
	ErrTimeOut        = errors.New("time out")
	ErrActionNotFound = errors.New("action not found")
)

//控制的接口
type CTRLOR interface {

	//GetCtrlTopic 获取ctrl的topic
	GetCtrlTopic(tid string, serviceID string, protocolPath string) (string, string, bool)
}

var ctrl CTRLOR

//SetCtrlRouter 设置控制路由处理模块
func SetCtrlRouter(ctrlSet CTRLOR) {
	ctrl = ctrlSet
}

//PubMsg 发布消息
func PubMsg(isP2P bool, topic string, obj interface{}) {
	if isP2P {
		innermq.P2PMsgPub(topic, obj)
	} else {
		innermq.MsgPub(topic, obj)
	}
}

//SubMs 订阅消息
func SubMsg(isP2P bool, topic string, obj interface{}, fun func(t string, b interface{})) {
	if isP2P {
		innermq.P2PMsgSub(topic, obj, fun)
	} else {
		innermq.MsgSub(topic, obj, fun)
	}
}

//SubAlarm 订阅告警广播，此时告警已经处理完成，框架向所有模块广播告警
func SubAlarm(fun func(curAlarm *eventtype.Alarm)) {
	SubMsg(false, "alarm", &eventtype.Alarm{}, func(t string, b interface{}) {
		palarm, ok := b.(*eventtype.Alarm)
		if ok {
			fun(palarm)
		}
	})
}

//RawAlarm 产生原始告警
func RawAlarm(curAlarm *eventtype.Alarm) {
	PubMsg(true, "raw_alarm", curAlarm)
}

//SubRawAlarm 订阅告警广播，此时告警已经处理完成，框架向所有模块广播告警
func SubRawAlarm(fun func(curAlarm *eventtype.Alarm)) {
	SubMsg(true, "raw_alarm", &eventtype.Alarm{}, func(t string, b interface{}) {
		palarm, ok := b.(*eventtype.Alarm)
		if ok {
			fun(palarm)
		}
	})
}

//RawData 产生数据 此时产生的数据为原始数据，设备上报的数据，需要dm处理后才能成为正式数据
//pid为""时，value必须为struct，此时pid通过tag来获取
func RawData(tid string, pid string, subid string, index int, value interface{},
	reportTime int64, pickTIme int64) {
	pValue := &propertytype.Value{
		TID:        tid,
		PID:        pid,
		SubID:      subid,
		Mindex:     index,
		Value:      value,
		ReportTime: reportTime,
		PickTime:   pickTIme,
	}
	PubMsg(true, "raw_data", pValue)
}

func RawDataMap(pData map[string]interface{}) {
	tid, _ := pData["tid"].(string)
	pid, _ := pData["pid"].(string)

	sid, _ := pData["sub_id"].(string)
	mmi, oki := pData["mindex"]
	//fmt.Println("data comming-----222222222-------", mmi, reflect.TypeOf(mmi))
	var mi = 0
	if oki {
		mf, okf := mmi.(float64)
		if okf {
			mi = int(mf)
		} else {
			mf32, okf32 := mmi.(float32)
			if okf32 {
				mi = int(mf32)
			} else {
				mi, oki = mmi.(int)
				if !oki {
					i64mi, ok64 := mmi.(int64)
					if ok64 {
						mi = int(i64mi)
					}
				}
			}
		}
	}
	rt, _ := pData["rt"].(int64)
	pt, _ := pData["pt"].(int64)

	RawData(tid, pid, sid, mi, pData["v"], rt, pt)
}

//RawDataEx 发送数据（指针）
func RawDataEx(value *propertytype.Value) {
	value.Value = propertytype.FmtValue(value.Value)
	PubMsg(true, "raw_data", value)
}

//SubRawData 订阅数据
func SubRawData(fun func(pValue *propertytype.Value)) {
	//PubMsg(true, "raw_data", pValue)
	SubMsg(true, "raw_data", &propertytype.Value{}, func(t string, b interface{}) {
		pvalue, ok := b.(*propertytype.Value)
		if ok {
			fun(pvalue)
		}
	})
}

//CTRCALLBACK 控制回调
type CTRCALLBACK func(pCmd *ctrltype.CtrlCmd) (map[string]interface{}, error)

func sendResult(pCmd *ctrltype.CtrlCmd, result map[string]interface{}, err error) {
	pResult := &ctrltype.CtrlResult{
		ID:     pCmd.ID,
		Seq:    pCmd.Seq,
		Result: result,
		Error:  err,
		//ModelID:modelID
	}
	id := "cmdresult." + pCmd.ID + "." + fmt.Sprintf("%d", pCmd.Seq)
	fmt.Println("sending----*************-", id, "^^^^^^^^^^^^^^^^", pResult.Result, pResult.Error)
	PubMsg(true, id, pResult)
	innermq.ExpiredMsgPub(id, time.Minute)

}

//PrintStack 打印堆栈
func PrintStack() {
	var buf [4096]byte
	n := runtime.Stack(buf[:], false)
	//fmt.Printf("==> %s\n", string(buf[:n]))
	logger.SLog.Error("panic !!!!", "stack", string(buf[:n]))
}

//SubCtrlCmd 订阅控制命令，处理控制命令，推荐直接使用iot-base/common/ctrl.SubCtrlCmd
//否则需要引用 _ "iot-base/common/ctrl"
func SubCtrlCmd(protocolName string, fun CTRCALLBACK) {

	//发送广播，如果接入运行了多个，则会出现处理多次的问题，后续会增加一个控制路由模块
	SubMsg(true, protocolName+".ctrl", &ctrltype.CtrlCmd{}, func(t string, cmd interface{}) {
		pCmd, ok := cmd.(*ctrltype.CtrlCmd)
		if ok {
			defer func() {
				// 发生宕机时，获取panic传递的上下文并打印

				err := recover()
				if err != nil {
					logger.SLog.Error("-------handle ctrlcmd error----panic------", "cmdid", pCmd.ID,
						"params", pCmd.Params, "error", err)
					PrintStack()
					switch err.(type) {
					case runtime.Error: // 运行时错误
						fmt.Println("@@@@@@@@@@@runtime error:", err)

					default: // 非运行时错误
						fmt.Println("error:", err)
					}

					sendResult(pCmd, nil, errors.New(fmt.Sprintln("error:", err)))
				}

			}()
			fmt.Println("------------call fun to server--------", pCmd.ID, protocolName)
			result, err := fun(pCmd)
			fmt.Println("------------call fun to server--------", pCmd.ID, result, err, protocolName)
			sendResult(pCmd, result, err)
		}
	})
}

func waitctrlcmd(pCmd *ctrltype.CtrlCmd, timeout time.Duration, fun CTRRESCALLBACK) {
	id := "cmdresult." + pCmd.ID + "." + fmt.Sprintf("%d", pCmd.Seq)
	//fmt.Println("waitformsg--------0000----*************-", id, "*********")
	innermq.WaitforMsg(id, &ctrltype.CtrlResult{}, timeout, func(t string, body interface{}) {
		result, ok := body.(*ctrltype.CtrlResult)
		if ok {
			if fun != nil {
				fun(result)
			}

		} else {
			fmt.Println("******waitctrlcmd************data type error**********", body)
		}
	}, func(topic string) {

		if fun != nil {
			pResult := &ctrltype.CtrlResult{
				ID:    pCmd.ID,
				Seq:   pCmd.Seq,
				Error: errors.New("time out"),
			}
			fun(pResult)
		}
	})
}

//CTRRESCALLBACK 结果回调
type CTRRESCALLBACK func(pResult *ctrltype.CtrlResult)

//Ctrl 控制
func Ctrl(pCmd *ctrltype.CtrlCmd, timeout time.Duration, fun CTRRESCALLBACK, isWait bool) error {
	/*tid := pCmd.ThingID

	pInfo, err := cache.GetThing(tid)
	if err != nil {
		return err
	}
	if pInfo == nil {
		return errors.New("device is not found")
	}*/
	//v, _ := uuid.NewV4()
	pCmd.Seq, _ = common.GetSnowflakeId()
	var ctopic = "thing.ctrl"
	if ctrl != nil {
		topic, edgeID, isCanCtrl := ctrl.GetCtrlTopic(pCmd.ThingID, pCmd.ID, pCmd.ProtocolPath)
		if topic != "" && isCanCtrl {
			ctopic = topic
			pCmd.EdgeID = edgeID
		}
		fmt.Println("------------get ctrl topic-------", ctopic, edgeID, isCanCtrl, pCmd.ProtocolPath)
	}
	PubMsg(true, ctopic, pCmd)
	fmt.Println("------------send ctrl topic-------", ctopic)
	if isWait {
		wg := sync.WaitGroup{}
		wg.Add(1)
		waitctrlcmd(pCmd, timeout, func(pResult *ctrltype.CtrlResult) {
			fun(pResult)
			wg.Done()
		})
		wg.Wait()
	} else {
		go waitctrlcmd(pCmd, timeout, fun)
	}
	//msgId = pInfo.ModelID + ".ctrlcmd."+pCmd.ID

	return nil
}

type AlarmSync struct {
	Alarms []*eventtype.Alarm
	IDS    []string
}

//ReportAllAlarms 上报所有有效告警，框架将针对这些告警进行告警同步
func ReportAllAlarms(alarms []*eventtype.Alarm, ids []string) {
	PubMsg(true, "alarms.report.cmds", AlarmSync{
		Alarms: alarms,
		IDS:    ids,
	})
}

//SubAllAlarmReport 订阅告警同步消息
func SubAllAlarmReport(fun func(ids []string, alarms []*eventtype.Alarm)) {
	//var alarms []*eventtype.Alarm
	var alarmInfo AlarmSync
	SubMsg(true, "alarms.report.cmds", &alarmInfo, func(t string, body interface{}) {
		alarmAll, ok := body.(*AlarmSync)
		if ok && fun != nil {
			fun(alarmAll.IDS, alarmAll.Alarms)
		}
	})
}

//CtrlThing 控制或服务
func CtrlThing(tid string, serviceName string, timeoutAsSecond int, params map[string]interface{}, fun CTRRESCALLBACK, isWait bool) error {
	pCmd := &ctrltype.CtrlCmd{
		ID:      serviceName,
		ThingID: tid,
		Params:  params,
	}
	return Ctrl(pCmd, time.Duration(timeoutAsSecond)*time.Second, fun, isWait)
}

const eventList = "rawEvent.msg"

//RawEvent 事件上报
func RawEvent(tid string, evid string, time time.Time,
	params map[string]interface{}, values []propertytype.Value) {
	var eventObj eventtype.Event
	eventObj.TID = tid
	eventObj.EventID = evid
	eventObj.Time = time
	eventObj.Params = params
	eventObj.Properties = values
	PubMsg(true, eventList, &eventObj)

}

//RawEventEx 事件上报
func RawEventEx(pEvent *eventtype.Event) {

	PubMsg(true, eventList, pEvent)

}

//SubEvent 订阅事件消息
func SubRawEvent(fun func(pEvent *eventtype.Event)) {
	SubMsg(true, eventList, &eventtype.Event{}, func(t string, b interface{}) {
		pEvent, ok := b.(*eventtype.Event)
		if ok {
			fun(pEvent)
		}
	})
}

/*

//SubDataReady 订阅数据
func SubEventReady(fun func(pValue *propertytype.Value)) {
	SubMsg(false, "event_ready", &propertytype.Value{}, func(t string, b interface{}) {
		pvalue, ok := b.(*propertytype.Value)
		if ok {
			fun(pvalue)
		}
	})
}

//EventReady 数据发布
func EventReady(value interface{}) {
	PubMsg(false, "event_ready", value)
}*/

//DataReady 数据发布
func DataReady(value interface{}) {
	PubMsg(false, "data_ready", value)
}

//SubDataReady 订阅数据
func SubDataReady(fun func(pValue *params.ValueInRedis)) {
	SubMsg(false, "data_ready", &params.ValueInRedis{}, func(t string, b interface{}) {
		pvalue, ok := b.(*params.ValueInRedis)
		if ok {
			fun(pvalue)
		}
	})
}

func CtrlResp(cmd *ctrltype.CtrlCmd) (map[string]interface{}, error) {
	logger.SLog.Infof("CtrlResp cmd %v", cmd)
	var pResult *ctrltype.CtrlResult
	Ctrl(cmd, 3*time.Second, func(pReply *ctrltype.CtrlResult) {
		pResult = pReply

	}, true)
	if pResult == nil {
		return nil, nil
	}
	if pResult.Error != nil && pResult.Error.Error() == "time out" {
		logger.SLog.Error("CtrlResp", "cmd", cmd, "result", pResult.Error)
		return nil, errors.New("time out")
	}
	if pResult.Error != nil {
		logger.SLog.Error("CtrlResp", "cmd", cmd, "result", pResult.Error)
		return nil, pResult.Error
	}
	return pResult.Result, nil
}

// EdgeReportData 上报数据(边缘处使用)
func EdgeReportData(isSync bool, action string, data []byte) (types.RespRecordData, error) {
	var respData types.RespRecordData
	if config.MQTTOption.IsMqttConnection == false {
		return respData, nil
	}
	seq, _ := common.GetSnowflakeId()
	req := types.ReqRecordData{
		ID:     action,
		Seq:    seq,
		Params: data,
	}
	PubMsg(true, "ReportData", req)
	// 等待回复结果
	if isSync {
		var buf strings.Builder
		buf.WriteString("reportDataCmd.")
		buf.WriteString(req.ID)
		buf.WriteString(strconv.FormatInt(req.Seq, 10))
		buf.WriteString(".msg.list")
		err := GetResult(buf.String(), &respData, 3)
		if err != nil {
			return respData, err
		}
		return respData, nil
	}
	return respData, nil
}

// CloudSubReportData 订阅上报消息(云端使用)
func CloudSubReportData(fun func(params *types.ReqRecordData) ([]byte, error)) {
	if fun != nil {
		// 订阅 redis 消息
		SubMsg(false, "ReportData", &types.ReqRecordData{}, func(topic string, obj interface{}) {
			if v, ok := obj.(*types.ReqRecordData); ok {

				// 执行函数
				result, err := fun(v)
				if errors.Is(err, ErrActionNotFound) {
					return
				}
				var buf strings.Builder
				buf.WriteString("cloud_report.")
				buf.WriteString(v.ID)
				buf.WriteString(strconv.FormatInt(v.Seq, 10))
				buf.WriteString(".msg.list")
				resp := types.RespRecordData{
					ID:  v.ID,
					Seq: v.Seq,
				}
				if err != nil {
					resp.Code = -1
					resp.Msg = err.Error()
				} else {
					// 成功消息
					resp.Code = 0
					resp.Result = result
				}
				//b, _ := json.Marshal(resp)
				//err = redis.RPushTimeOut(buf.String(), b, 3)
				err = PushAsyMsg(buf.String(), resp, 3)
			}
		})
	}
}

// CloudSendCmd 云端发送命令.
func CloudSendCmd(isSync bool, edgeId, action string, data []byte) (types.RespActionData, error) {
	var respData types.RespActionData
	if config.MQTTOption.IsMqttConnection == false {
		return respData, nil
	}
	seq, _ := common.GetSnowflakeId()
	req := types.ReqActionData{
		Action:  action,
		Seq:     seq,
		Params:  data,
		EdgeId:  edgeId,
		SeqTime: time.Now(),
	}
	PubMsg(true, "SendActionCmd", req)
	// 等待回复结果
	if isSync {
		var buf strings.Builder
		buf.WriteString("cloud_SendActionCmd.")
		buf.WriteString(req.Action)
		buf.WriteString(strconv.FormatInt(req.Seq, 10))
		buf.WriteString(".msg.list")
		err := GetResult(buf.String(), &respData, 3)
		if err != nil {
			return respData, err
		}
		return respData, nil
	}
	return respData, nil
}

// EdgeSubActionCmd 边缘测订阅命令.
func EdgeSubActionCmd(fun func(params *types.ReqActionData) ([]byte, error)) {
	if fun != nil {
		// 订阅 redis 消息
		SubMsg(false, "SendActionCmd", &types.ReqActionData{}, func(topic string, obj interface{}) {
			if v, ok := obj.(*types.ReqActionData); ok {
				// 执行函数
				result, err := fun(v)
				if errors.Is(err, ErrActionNotFound) {
					return
				}
				resp := types.RespActionData{
					Seq:    v.Seq,
					Action: v.Action,
				}
				if err != nil {
					resp.Code = -1
					resp.Msg = err.Error()
				} else {
					// 成功消息
					resp.Code = 0
					resp.Result = result
				}
				var buf strings.Builder
				buf.WriteString("edge_SendActionCmd.")
				buf.WriteString(v.Action)
				buf.WriteString(strconv.FormatInt(v.Seq, 10))
				buf.WriteString(".msg.list")
				err = PushAsyMsg(buf.String(), resp, 3)
			}
		})
	}
}

func GetResult(topic string, result interface{}, exp int) error {
	resp, err := redis.BLPop(exp, topic)
	if err == redis2.Nil {
		return ErrTimeOut //超时
	}
	if err != nil {
		return err
	}
	if len(resp) == 0 && err == nil {
		return ErrTimeOut //超时
	}
	err = msgpack.Unmarshal([]byte(resp[1]), result)
	return err
}

func PushAsyMsg(topic string, param interface{}, exp int) error {
	b, err := msgpack.Marshal(param)
	err = redis.RPushTimeOut(topic, b, exp)
	return err
}
